package com.jerry.question;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;

import java.io.*;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * Created with IntelliJ IDEA
 * User: Jerry
 * Date: 2020/8/6
 * Time: 17:32
 * Description: 启动参数：-Xmx128m
 */
public class Question {

    public static void main(String[] args) throws Exception {
        System.out.println("开始时间：" + new Date());
        long start = System.currentTimeMillis();

        Question bigDataRead = new Question();
        bigDataRead.execute();

        System.out.println("结束时间：" + new Date());
        System.out.println("全部执行完毕。耗时：" + (System.currentTimeMillis() - start) / 1000 + "秒");
    }

    /**
     * 拆分并入库
     *
     * @throws Exception
     */
    public void execute() throws Exception {
        // 拆分文件
        long start = System.currentTimeMillis();
        System.out.println("开始拆分文件...");
        int fileCount = splitFile();
        System.out.println("文件拆分排序完毕，一共拆分了" + fileCount + "个文件。耗时：" + (System.currentTimeMillis() - start) / 1000 + "秒");

        // 插入数据库
        long insertStart = System.currentTimeMillis();
        System.out.println("开始录入数据库...");
        insertDb(fileCount);
        System.out.println("数据库录入完毕。耗时：" + (System.currentTimeMillis() - insertStart) / 1000 + "秒");
    }

    /**
     * 拆分文件
     *
     * @throws IOException
     */
    private int splitFile() throws Exception {
        // 文件最大行
        int fileMaxLine = 600000;
        // 记录文件个数
        int fileCount = 0;
        // 记录当前已保存行数
        int lineCount = 0;
        // 每行数据
        String line;
        // 作为heap的构造参数用
        List<PtBigdata> list = new ArrayList<>();

        FileReader read = new FileReader("testjson.txt");
        BufferedReader br = new BufferedReader(read);
        while (StringUtils.isNotBlank((line = br.readLine()))) {
            lineCount++;
            PtBigdata ptBigdata = JSONObject.parseObject(line, PtBigdata.class);
            list.add(ptBigdata);

            // 达到最大行数就新写一个文件
            if (lineCount == fileMaxLine) {
                writeFile(list, fileCount);
                fileCount++;
                lineCount = 0;
            }
        }

        if (!list.isEmpty()) {
            writeFile(list, fileCount);
            fileCount++;
        }

        br.close();

        return fileCount;
    }

    /**
     * 数据写入小文件
     *
     * @param list
     * @param fileCount
     * @throws Exception
     */
    private void writeFile(List<PtBigdata> list, int fileCount) throws Exception {
        // 排序用最大堆
        MaxHeap<PtBigdata> heap = new MaxHeap<>(list.toArray(new PtBigdata[0]));

        String fileName = "temp" + fileCount + ".txt";
        BufferedWriter bw = new BufferedWriter(new FileWriter(fileName));
        while (!heap.isEmpty()) {
            PtBigdata data = heap.extractMax();
            String s = JSONObject.toJSONString(data);

            bw.write(s);
            bw.newLine();
        }
        bw.close();

        list.clear();
    }

    /**
     * 插入数据库
     *
     * @param fileCount
     */
    private void insertDb(int fileCount) throws Exception {
        MaxHeap<PtBigdata> heap = new MaxHeap<>(fileCount);
        List<PtBigdata> list = new ArrayList<>();

        Connection conn = jdbcInit();
        if (conn == null) {
            System.out.println("获取连接失败");
            return;
        }

        BufferedReader[] bufferedReaders = new BufferedReader[fileCount];
        for (int i = 0; i < fileCount; i++) {
            FileReader fileReader = new FileReader("temp" + i + ".txt");
            bufferedReaders[i] = new BufferedReader(fileReader);
            readData(heap, bufferedReaders[i], i);
        }

        // 插入的总行数
        int totalCount = 0;
        // 批量插入数据库的行数
        int patchCount = 10000;
        while (!heap.isEmpty()) {
            PtBigdata ptBigdata = heap.extractMax();
            list.add(ptBigdata);
            if (list.size() == patchCount) {
                jdbcInsert(list, conn);
                totalCount += patchCount;
                if (totalCount % 10000000 == 0) {
                    System.out.println("已经插入了" + totalCount + "行数据");
                }
                list.clear();
            }

            Integer index = ptBigdata.getIndex();
            readData(heap, bufferedReaders[index], index);
        }

        for (BufferedReader bufferedReader : bufferedReaders) {
            bufferedReader.close();
        }

        conn.close();
    }

    /**
     * 读取数据放入堆中
     *
     * @param heap
     * @param bufferedReader
     * @param i
     * @throws IOException
     */
    private void readData(MaxHeap<PtBigdata> heap, BufferedReader bufferedReader, int i) throws IOException {
        String line = bufferedReader.readLine();
        if (line != null) {
            PtBigdata ptBigdata = JSONObject.parseObject(line, PtBigdata.class);
            ptBigdata.setIndex(i);
            heap.add(ptBigdata);
        }
    }

    /**
     * 初始化数据库连接
     *
     * @return
     * @throws Exception
     */
    private Connection jdbcInit() throws Exception {
        String name = "com.mysql.jdbc.Driver";
        // 数据库
        String url = "jdbc:mysql://localhost:3306/question?useUnicode=true&characterEncoding=utf-8&useSSL=false&zeroDateTimeBehavior=convertToNull";
        String user = "root";
        String password = "123456";

        Class.forName(name);
        Connection conn = DriverManager.getConnection(url, user, password);
        if (conn != null) {
            System.out.println("获取连接成功");
            return conn;
        }
        return null;
    }

    /**
     * 批量插入数据
     *
     * @param list
     * @param conn
     */
    private void jdbcInsert(List<PtBigdata> list, Connection conn) {
        // sql前缀
        String prefix = "INSERT INTO pt_bigdata (number,name) VALUE ";
        try {
            // 保存sql后缀
            StringBuilder suffix = new StringBuilder();
            conn.setAutoCommit(false);
            // 准备执行语句
            PreparedStatement pst = conn.prepareStatement("");
            // 循环拼凑
            for (PtBigdata ptBigdata : list) {
                suffix.append("(\"").append(ptBigdata.getNumber()).append("\",\"")
                        .append(ptBigdata.getName()).append("\"),");
            }
            // 构建完整SQL
            String sql = prefix + suffix.substring(0, suffix.length() - 1);
            // 添加批量执行SQL
            pst.addBatch(sql);
            // 执行操作
            pst.executeBatch();
            // 提交事务
            conn.commit();
            pst.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    /**
     * 验证所有文件加起来行数是否等于一亿
     *
     * @param count
     * @throws Exception
     */
    private void countLineTotal(int count) throws Exception {
        int line = 0;
        for (int i = 0; i < count; i++) {
            FileReader fileReader = new FileReader("temp" + i + ".txt");
            BufferedReader bufferedReader = new BufferedReader(fileReader);
            while (bufferedReader.readLine() != null) {
                line++;
            }
        }
        System.out.println("总行数：" + line);
    }

}